package club.monkeywood.ad.dmp.graphx

import java.io

import org.apache.spark.SparkConf
import org.apache.spark.graphx.{Edge, Graph, VertexId}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

import scala.collection.immutable


/**
	* 若用户有多个userId，将同一个用户的多个userId数据通过连通图合并成一条
	* 输入：文本数据
	* 张三  Angelababy  波多  APP爱奇艺:10   BS汉阳门:8
	* 波多  老赵  K天龙八部:50
	* 老段  老赵  老羊  APP爱奇艺:10
	* 刘亦菲 K三生三世十里桃花:5
	* 输出：
	*/
object UserTagsMergeDemo{

	def main(args: Array[String]): Unit = {

		// 0 校验参数个数
		if (args.length != 1) {
			println(
				"""
					|cn.dmp.tools.UserTagsMergeDemo
					|参数：
					| userInputPath
				""".stripMargin)
			sys.exit()
		}

		// 1 接受程序参数
		val Array(userInputPath) = args

		// 2 创建sparkconf->SparkSession
		val sparkConf = new SparkConf()
		sparkConf.setAppName(s"${this.getClass.getSimpleName}")
		sparkConf.setMaster("local[*]")
		//使用KryoSerializer更快
		// RDD 序列化到磁盘 worker与worker之间的数据传输
		sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
		//避免字段名过长报错
		sparkConf.set("spark.debug.maxToStringFields", "100")
		val ss = SparkSession.builder()
			.appName(s"${this.getClass.getSimpleName}")
			.config(sparkConf)
			.getOrCreate()
		val sc = ss.sparkContext

		/**
			* 输出：
			* Array[ Array[张三,Angelababy,波多,APP爱奇艺:10,BS汉阳门:8],
			* Array[波多,老赵,K天龙八部:50],
			* Array[老段,老赵,老羊,APP爱奇艺:10],
			* Array[刘亦菲,K三生三世十里桃花:5] ]
			*/
		val data: RDD[Array[String]] = sc.textFile(userInputPath)
			.map(_.split(","))

		/**
			* 根据读取数据创建点集合
			* 注意：顶点数据可以重复，即相同的顶点id可以出现多条，但顶点带的属性不同
			* 输出：
			* uv = {
			*   (张三.hashCode.ToLong, (张三, List(APP爱奇艺:10, BS汉阳门:8))),
			*   (Angelababy.hashCode.ToLong, (Angelababy, List())),
			*   (波多.hashCode.ToLong, (波多, List())),
			*
			*   (波多.hashCode.ToLong, (波多, List(K天龙八部:50))),
			*   (老赵.hashCode.ToLong, (老赵, List())),
			*
			*   (老段.hashCode.ToLong, (老段, List(APP爱奇艺:10))),
			*   (老赵.hashCode.ToLong, (老赵, List())),
			*   (老羊.hashCode.ToLong, (老羊, List())),
			*
			*   (刘亦菲.hashCode.ToLong, (刘亦菲, List(K三生三世十里桃花:5)))
			* }
			*
			* (774889,(张三,List(APP爱奇艺:10, BS汉阳门:8)))
			* (608971126,(Angelababy,List()))
			* (886904,(波多,List()))
			*
			* (886904,(波多,List(K天龙八部:50)))
			* (1052052,(老赵,List()))
			*
			* (1043412,(老段,List(APP爱奇艺:10)))
			* (1052052,(老赵,List()))
			* (1048489,(老羊,List()))
			*
			* (20854308,(刘亦菲,List(K三生三世十里桃花:5)))
			*/
		val uv: RDD[(Long, (String, List[(String, Int)]))] = data.flatMap(arr => {
			val userNames: Array[String] = arr.filter(_.indexOf(":") == -1)
			val userTags: List[(String, Int)] = arr.filter(_.indexOf(":") != -1)
					.map(tag => {
						val ts: Array[String] = tag.split(":")
						(ts(0), ts(1).toInt)
					})
  				.toList

			//只让第一个元素携带标签数据
			userNames.map(name => {
				if(name.equals(userNames(0))) {
					(name.hashCode.toLong, (name, userTags))
				}else {
					(name.hashCode.toLong, (name, List.empty[(String, Int)]))
				}
			})

		})

		println("打印uv:--------------------------------------------------------")
		/**
			* (774889,(张三,List(APP爱奇艺:10, BS汉阳门:8)))
			* (1043412,(老段,List(APP爱奇艺:10)))
			* (608971126,(Angelababy,List(APP爱奇艺:10, BS汉阳门:8)))
			* (1052052,(老赵,List(APP爱奇艺:10)))
			* (886904,(波多,List(APP爱奇艺:10, BS汉阳门:8)))
			* (1048489,(老羊,List(APP爱奇艺:10)))
			* (886904,(波多,List(K天龙八部:50)))
			* (20854308,(刘亦菲,List(K三生三世十里桃花:5)))
			* (1052052,(老赵,List(K天龙八部:50)))
			*/
		uv.foreach(println)

		/**
			* 创建边集合
			* 输出：
			* ue = {
			*   (张三.hashCode.ToLong, Angelababy.hashCode.ToLong, 0),
			*   (张三.hashCode.ToLong, Angelababy.hashCode.ToLong, 0),
			*   (张三.hashCode.ToLong, 波多.hashCode.ToLong, 0),
			*
			*   (波多.hashCode.ToLong, 波多.hashCode.ToLong, 0),
			*   (波多.hashCode.ToLong, 老赵.hashCode.ToLong, 0),
			*
			*   (老段.hashCode.ToLong, 老段.hashCode.ToLong, 0),
			*   (老段.hashCode.ToLong, 老赵.hashCode.ToLong, 0),
			*   (老段.hashCode.ToLong, 老羊.hashCode.ToLong, 0),
			*
			*   (刘亦菲.hashCode.ToLong, 刘亦菲.hashCode.ToLong, 0),
			* }
			*/
		val ue: RDD[Edge[Int]] = data.flatMap(arr=>{
			val userNames = arr.filter(_.indexOf(":") == -1)
			//0:表示边无方向
			userNames.map(name => Edge(userNames(0).hashCode.toLong, name.hashCode.toLong, 0))
		})

		//创建图
		val graph = Graph(uv, ue)
		val cc = graph.connectedComponents().vertices
		println("打印连通图:----------------------------------------------")
		/**
			* 一个顶点id只会出现在一行数据中
			* (774889,774889)
			* (1048489,774889)
			* (20854308,20854308)
			* (886904,774889)
			* (608971126,774889)
			* (1052052,774889)
			* (1043412,774889)
			*/
		cc.foreach(println)

		//聚合数据
		println("打印聚合数据：--------------------------------------------------------")
		/**
			* (20854308,(刘亦菲,List(K三生三世十里桃花:5)))
			* (774889,(波多,波多,Angelababy,老赵,老赵,老段,张三,老羊,List(APP爱奇艺:10, BS汉阳门:8, K天龙八部:50, APP爱奇艺:10, BS汉阳门:8, K天龙八部:50, APP爱奇艺:10, APP爱奇艺:10, APP爱奇艺:10, BS汉阳门:8, APP爱奇艺:10)))
			*/
		val mid1: RDD[(VertexId, (Set[String], List[(String, Int)]))] = cc.join(uv)
  			.map{
				  // Set(name):取出重复姓名
				  case (userId, (commonId, (name, tags)))=> (commonId, (Set(name), tags))
			  }
				//写法一：用普通函数
  		  //.reduceByKey((t1,t2)=>(t1._1++t1._1,t2._2++t2._2))
				//写法二：用偏函数
				.reduceByKey {
				  case ((name1, tags1),(name2, tags2)) => (name1++name2 ,tags1++tags2)
				}

		val ret: RDD[(Set[String], List[(String, Int)])] = mid1.map{
	    case (userId, (names, tags)) => (names, tags.groupBy(_._1)
		                                    .mapValues(_.foldLeft(0)(_+_._2))
	                                      .toList)
    }

    ret.foreach(println)

		ss.close()
	}

}
